package com.atguigu.flink.streamx;

import com.streamxhub.streamx.flink.core.StreamEnvConfig;
import com.streamxhub.streamx.flink.core.java.function.SQLFromFunction;
import com.streamxhub.streamx.flink.core.java.sink.JdbcSink;
import com.streamxhub.streamx.flink.core.java.source.KafkaSource;
import com.streamxhub.streamx.flink.core.scala.StreamingContext;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 */
public class StreamxMySQLSinkDemo {
    public static void main(String[] args) {

        StreamEnvConfig envConfig = new StreamEnvConfig(args, null);
        StreamingContext context = new StreamingContext(envConfig);

        SingleOutputStreamOperator<WaterSensor> kafkaDS = new KafkaSource<String>(context)
                .alias("kafka1")
                .topic("s1")
                .getDataStream()
                .map(record -> {
                    String[] split = record.value().split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[1]));
                });

        new JdbcSink<WaterSensor>(context)
                .sql(new SQLFromFunction<WaterSensor>() {
                    @Override
                    public String from(WaterSensor waterSensor) {
                        return String.format(
                                "insert into sensor(id,ts,vc) values('%s',%d,%d)",
                                waterSensor.getId(), waterSensor.getTs(), waterSensor.getVc());
                    }
                })
                .sink(kafkaDS);


        context.start();


    }
}
